package com.diy.sigmund.diyorder.common.middleware.mq;

import com.diy.sigmund.diycommon.entity.base.exception.DiyRuntimeException;
import com.diy.sigmund.diycommon.test.entity.MqTestVO;
import com.diy.sigmund.diycommon.util.json.JacksonUtil;
import com.diy.sigmund.diyorder.common.middleware.mq.rocketmq.spring.one.ProductWithPayload;
import com.diy.sigmund.diyorder.common.spring.config.CustomRocketMQConfig;
import com.fasterxml.jackson.core.type.TypeReference;
import java.util.List;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author snowalker
 * @version 1.0
 * @date 2019/6/15 11:16
 * @className SecKillChargeOrderConsumer
 * @desc 秒杀下单消费者
 */
@Component
public class SecKillChargeOrderConsumer {

    private static final Logger LOG = LoggerFactory.getLogger(SecKillChargeOrderConsumer.class);

    @Autowired
    CustomRocketMQConfig customRocketMQConfig;

    @Resource(name = "secKillChargeOrderListenerImpl")
    private MessageListenerConcurrently messageListener;
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * ConsumeConcurrentlyStatus.RECONSUME_LATER :
     * consumer 消费失败，会把消息重新发往 %RETRY% + consumerGroup，这个 retry 消息会在一定时间后，真实送到retry topic
     */
    @PostConstruct
    public void init() {
        DefaultMQPushConsumer defaultMQPushConsumer = new DefaultMQPushConsumer(
                RocketMQProtocolEnum.SECKILL_TEST_TOPIC.getConsumerGroup());
        defaultMQPushConsumer.setNamesrvAddr(customRocketMQConfig.getNameSrvAddr());
        // 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
        // 有三个消费点: CONSUME_FROM_LAST_OFFSET:消费者客户端从之前停止的地方开始。
        // 如果是新起步的消费客户端，根据消费群体的老化情况，
        // 有两种情况: 如果消费者组是最近创建的，以至于订阅的最早的消息还没有过期，这意味着消费者组代表最近启动的业务，则消费将从头开始;
        // 如果订阅的最早的消息已经过期，消费将从最新的消息开始，这意味着在引导时间戳之前生成的消息将被忽略。
        // CONSUME_FROM_FIRST_OFFSET:使用者客户端将从最早可用的消息开始。
        // CONSUME_FROM_TIMESTAMP:使用者客户端将从指定的时间戳开始，这意味着在consumeTimestamp之前生成的消息将被忽略
        defaultMQPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        // 消费模式:集群模式
        // 集群：同一条消息 只会被一个消费者节点消费到
        // 广播：同一条消息 每个消费者都会消费到
        defaultMQPushConsumer.setMessageModel(MessageModel.CLUSTERING);
        // 设置每次拉取的消息量，默认为1；消息量很大（千，万），才能看到效果；批量消息不支持延迟消息
        defaultMQPushConsumer.setConsumeMessageBatchMaxSize(1);
        // 注册监听器
        // defaultMQPushConsumer.registerMessageListener(messageListener);
        defaultMQPushConsumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                try {
                    // TimeUnit.SECONDS.sleep(2000);
                    TimeUnit.MILLISECONDS.sleep(2);
                } catch (InterruptedException e) {
                    LOG.error("", e);
                }
                for (MessageExt msg : msgs) {
                    // LOG.info("顺序消费msg:{}", msg);
                    // LOG.info("body：{}", new String(msg.getBody()));
                    LOG.info("body：{}", new String(msg.getBody()));
                    // SendResult sendResult = rocketMQTemplate.syncSend("FAST_PROCESSING_TOPIC:*", new String(msg.getBody()));
                    ProductWithPayload<String> sendResult = rocketMQTemplate.sendAndReceive("FAST_PROCESSING_TOPIC:*",
                            new String(msg.getBody()),
                            new TypeReference<ProductWithPayload<String>>() {
                            }.getType(),
                            30000,
                            0);
                    // SendResult sendResult = rocketMQTemplate.syncSend("FAST_PROCESSING_TOPIC:*",
                    //         JacksonUtil.toObject(new String(msg.getBody()), MqTestVO.class));
                    LOG.info("顺序消费msg:{}", msg);
                    LOG.info("顺序消费sendResult:{}", sendResult);
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        // defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
        //     @Override
        //     public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        //         try {
        //             TimeUnit.MILLISECONDS.sleep(2);
        //         } catch (InterruptedException e) {
        //             LOG.error("", e);
        //         }
        //         for (MessageExt msg : msgs) {
        //             LOG.info("body：{}", new String(msg.getBody()));
        //             SendResult sendResult = rocketMQTemplate.syncSend("FAST_PROCESSING_TOPIC:*", new String(msg.getBody()));
        //             LOG.info("并发消费msg:{}", msg);
        //             LOG.info("并发消费sendResult:{}", sendResult);
        //         }
        //         return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        //     }
        // });
        // 订阅所有消息
        try {
            // *表示不过滤，可以通过tag来过滤
            defaultMQPushConsumer.subscribe(RocketMQProtocolEnum.SECKILL_TEST_TOPIC.getTopic(), "*");
            // 启动消费者
            defaultMQPushConsumer.start();
        } catch (MQClientException e) {
            LOG.error("[秒杀下单消费者]--SecKillChargeOrderConsumer加载异常", e);
            throw new DiyRuntimeException("[秒杀下单消费者]--SecKillChargeOrderConsumer加载异常!");
        }
        LOG.info("[秒杀下单消费者]--SecKillChargeOrderConsumer加载完成!");
    }
}
